package com.shujia.spark.sql.test

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{Column, DataFrame, Row, SparkSession}

object Test01 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("test")
      .master("local")
      .getOrCreate()

    //读取test文件
    val burkDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("burk STRING,year STRING,tsl01 DOUBLE,tsl02 DOUBLE,tsl03 DOUBLE,tsl04 DOUBLE,tsl05 DOUBLE,tsl06 DOUBLE,tsl07 DOUBLE,tsl08 DOUBLE,tsl09 DOUBLE,tsl10 DOUBLE,tsl11 DOUBLE,tsl12 DOUBLE")
      .load("data/test.txt")

    //导入隐式转换
    import spark.implicits._

    //导入spark所有函数
    import org.apache.spark.sql.functions._
    /**
      * 1、统计每个公司每年按月累计收入
      *
      * 输出结果
      * 公司代码,年度,月份,当月收入,累计收入
      */

    burkDF.createOrReplaceTempView("burk")

    //map(1,tsl01,2,tsl02,3,tsl03,4,tsl04,5,tsl05,6,tsl06,7,tsl07,8,tsl08,9,tsl09,10,tsl10,11,tsl11,12,tsl12)
    spark.sql(
      """
        |select burk,year,mouth,plc,sum(plc) over (partition by burk,year order by mouth)as sumPlc
        |from(
        select burk,year,explode(map(1,tsl01,2,tsl02,3,tsl03,4,tsl04,5,tsl05,6,tsl06,7,tsl07,8,tsl08,9,tsl09,10,tsl10,11,tsl11,12,tsl12)) as (mouth,plc)
        from burk
        |) as c
      """.stripMargin).show()

    spark.sql(
      """
        |select burk,year,mouth,plc,sum(plc) over(partition by burk,year order by mouth)as sumPlc
        |from(
        |select burk,year,mouth,plc
        |from burk
        |lateral view explode(map(1,tsl01,2,tsl02,3,tsl03,4,tsl04,5,tsl05,6,tsl06,7,tsl07,8,tsl08,9,tsl09,10,tsl10,11,tsl11,12,tsl12)) t as mouth,plc
        |) as a
      """.stripMargin)

    /**
      * 使用UDL写
      */
    val mapData: Column = map(
      expr("1"), $"tsl01",
      expr("2"), $"tsl02",
      expr("3"), $"tsl03",
      expr("4"), $"tsl04",
      expr("5"), $"tsl05",
      expr("6"), $"tsl06",
      expr("7"), $"tsl07",
      expr("8"), $"tsl08",
      expr("9"), $"tsl09",
      expr("10"), $"tsl10",
      expr("11"), $"tsl11",
      expr("12"), $"tsl12"
    )

   burkDF.select($"burk",$"year",explode(mapData) as Array("mouth","plc"))
      .withColumn("sumPlc",sum($"plc") over Window.partitionBy($"burk",$"year").orderBy("mouth"))
//      .show(1000)


  }
}
